package com.minivision.fdiot.mqtt;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.minivision.fdiot.common.RestResult;
import com.minivision.fdiot.entity.Device;
import com.minivision.fdiot.service.DeviceService;
import com.minivision.fdiot.service.DiscoveryService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.moquette.connections.IConnectionsManager;
import io.moquette.server.Server;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.client.RestTemplate;

/**
 * MQTT连接管理器
 * @author hughzhao
 * @2018年3月2日
 */
@Component
@Slf4j
public class MqttConnectionManager {

  @Autowired
  private Server mqttBroker;

  @Autowired
  private DeviceService deviceService;

  @Autowired
  private DiscoveryService discoveryService;

  @Autowired
  private ObjectMapper mapper;

  @Autowired
  private RestTemplate restTemplate;

  /**
   * Description: 断开连接，集群部署时可能存在以下问题
   * 1.设备连接到该broker
   * 2.设备连接到其他的broker<br>
   *
   * @author guming <br>
   * @taskId
   * @param
   * @return <br>
   */
  public void closeConnection(String clientId) {
    IConnectionsManager manager = mqttBroker.getConnectionsManager();
    Device device = deviceService.findDevice(clientId);
    if(device.getOnline()){//当且仅当设备在线时，才能断开连接，如果设备已主动下线，则该处不处理
      if (manager.isConnected(clientId)) {
        manager.closeConnection(clientId, true);
        log.info("close connection for device[{}]", clientId);
      }else {//设备连接到其他的broker
        String node = device.getConnectNode();
        if(discoveryService.isUp(node)){
          String url = "http://" + node + "/api/v1/device/disConnectDevice";
          String json = null;
          try {
            json = mapper.writeValueAsString(device);
          } catch (JsonProcessingException e) {
            log.error("device type[" + device.getClass().getSimpleName() + "] can not be serialized to json", e);
          }
          try {
            RestResult<?> restResult = restTemplate.postForObject(url, device, RestResult.class);
            log.info("message[{}] to device[{}] forward to {}, response:{}", json, clientId, node, restResult.toString());
            if (restResult.getStatus() != 0) {
              log.error("can not forward to " + url + ", maybe " + node + " was down...");
            }
          } catch (Exception e) {
            log.error("can not forward to " + url + ", maybe " + node + " was down...", e);
          }
        }
      }
    }
  }

  /**
   * Description: 断开连接<br>
   *
   * @author guming <br>
   * @taskId
   * @param
   * @return <br>
   */
  public void closeLocalConnection(String clientId){
    IConnectionsManager manager = mqttBroker.getConnectionsManager();
    if (manager.isConnected(clientId)) {
      manager.closeConnection(clientId, true);
      log.info("close connection for device[{}]", clientId);
    }
  }
  
  public boolean isConnected(String clientId) {
    IConnectionsManager manager = mqttBroker.getConnectionsManager();
    return manager.isConnected(clientId);
  }
  
}
